package com.eyas.im.client.nio.handler;

import com.eyas.im.client.TaskCompleteFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

class SocketClientIOEventHandler {
    private final static Logger logger = LoggerFactory.getLogger(SocketClientIOEventHandler.class);
    private SelectionKey selectionKey;
    private SocketChannel socketChannel;
    private ByteBuffer readBuffer;
    private ByteBuffer writeBuffer;
    private ConcurrentSkipListMap<TaskCompleteFuture, byte[]> writeBytesCache = new ConcurrentSkipListMap<>();
    private Lock lock = new ReentrantLock();
    private TaskCompleteFuture currentWriteTaskFuture;

    public SocketClientIOEventHandler(int bufferCapacity) {
        this.readBuffer = ByteBuffer.allocate(bufferCapacity);
        this.writeBuffer = ByteBuffer.allocate(bufferCapacity);
    }

    final void bind(SelectionKey selectionKey, SocketChannel socketChannel) {
        this.selectionKey = selectionKey;
        this.socketChannel = socketChannel;
    }


    public byte[] read() throws IOException, ChannelCloseException {
        int readSize = socketChannel.read(readBuffer);
        if (readSize == -1) {
            //client端主动断开
            throw new ChannelCloseException();
        } else {
            readBuffer.flip();
            byte[] bytes = new byte[readSize];
            readBuffer.get(bytes, 0, readSize);
            readBuffer.clear();
            return bytes;
        }
    }

    public void write() throws IOException {
        try {
            if (this.writeBuffer.hasRemaining() && currentWriteTaskFuture != null) { //说明这个buffer上次write操作已经写了一部分了，但还没有写完
                //如果writebuffer还没写完，继续写数据
                this.socketChannel.write(this.writeBuffer);
                if (!this.writeBuffer.hasRemaining()) {
                    //如果writebuffer写结束，则调用TaskFuture的complete方法
                    this.currentWriteTaskFuture.complete(null);
                    this.currentWriteTaskFuture = null;
                    //判断是否还有未发送完的数据
                    Map.Entry<TaskCompleteFuture, byte[]> entry = writeBytesCache.pollFirstEntry();
                    if (entry == null) {
                        //数据都发送结束,这里加锁再提取一遍，如果还没有数据，则取消OP_WRITE事件
                        try {
                            lock.lock();
                            entry = writeBytesCache.pollFirstEntry();
                            if (entry == null) {
                                //可以确定截止到最后一次flush操作，所有的数据都发送完毕，此时可以取消OP_WRITE事件
                                this.maskNotWrite();
                                return;
                            }
                        } finally {
                            lock.unlock();
                        }
                    }

                    //走到这里，说明cache中还有未处理的数据
                    this.writeBuffer.clear();
                    this.writeBuffer.put(entry.getValue());
                    this.writeBuffer.flip();
                    this.currentWriteTaskFuture = entry.getKey();
                    //修改完后，即可终止程序，等待下次OP_WRITE事件时处理
                } else {
                    //本次的writebuffer还有没写完的数据，不做任何处理，等待下次OP_WRITE事件处理
                    logger.debug("数据没有完全写入socket buffer中，等待下次继续");
                }
            } else {
                //writeBuffer为空或者上一次flush的数据已经处理结束
                this.writeBuffer.clear();
                Map.Entry<TaskCompleteFuture, byte[]> entry = writeBytesCache.pollFirstEntry();
                if (entry == null) {
                    /*进入这个判断有两种情况
                     * 1、一次性将一个writeBuffer完整的写出到socket buffer中
                     * 2、上一次write操作进入IOException处理流程
                     * */
                    try {
                        lock.lock();
                        entry = writeBytesCache.pollFirstEntry();
                        if (entry == null) {
                            this.maskNotWrite();
                            return;
                        }
                    } finally {
                        lock.unlock();
                    }
                }

                //到这里，说明已经提取出了要发送的数据信息
                this.writeBuffer.put(entry.getValue());
                this.currentWriteTaskFuture = entry.getKey();
                this.writeBuffer.flip();
                this.socketChannel.write(this.writeBuffer);
                if (!this.writeBuffer.hasRemaining()) {
                    //writebuffer写结束，调用TaskFuture的complete方法
                    this.currentWriteTaskFuture.complete(null);
                    this.currentWriteTaskFuture = null;
                } else {
                    //本次的writebuffer还有没写完的数据，不做任何处理，等待下次OP_WRITE事件处理
                    logger.debug("数据没有完全写入socket buffer中，等待下次继续");
                }
            }
        } catch (IOException e) {
            //出现异常
            this.writeBuffer.clear();
            this.currentWriteTaskFuture.complete(e);
            this.currentWriteTaskFuture = null;
            throw e;
        }
    }


    void pushWriteBytes(TaskCompleteFuture taskCompleteFuture, byte[] bytes) {
        Objects.requireNonNull(taskCompleteFuture);
        Objects.requireNonNull(bytes);
        this.writeBytesCache.put(taskCompleteFuture, bytes);
    }

    private void maskNotWrite() {
        try {
            lock.lock();
            this.selectionKey.interestOps(this.selectionKey.interestOps() ^ SelectionKey.OP_WRITE);
        } finally {
            lock.unlock();
        }
    }

    void maskWrite() {
        try {
            lock.lock();
            this.selectionKey.interestOps(this.selectionKey.interestOps() | SelectionKey.OP_WRITE);
        } finally {
            lock.unlock();
        }
    }

    void shutdown0() throws IOException {
        this.selectionKey.cancel();
        this.socketChannel.close();
    }

}
